package com.atguigu.flink.chapter07.state;

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.state.*;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.runtime.state.FunctionInitializationContext;
import org.apache.flink.runtime.state.FunctionSnapshotContext;
import org.apache.flink.streaming.api.checkpoint.CheckpointedFunction;
import org.apache.flink.streaming.api.datastream.BroadcastStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.streaming.api.functions.co.BroadcastProcessFunction;
import org.apache.flink.util.Collector;

import java.util.ArrayList;
import java.util.Arrays;

public class Flink03_Operator_BroadCast {

    private static MapStateDescriptor<String, String> bcState;

    public static void main(String[] args) throws Exception {
        Configuration conf = new Configuration();
        conf.setInteger("rest.port", 20000);
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(conf);
        env.setParallelism(2);
        // 3s做一次快照
        env.enableCheckpointing(3000);

        DataStreamSource<String> dataStream = env.socketTextStream("hadoop162", 9999);
        DataStreamSource<String> controlStream = env.socketTextStream("hadoop162", 8888);

        MapStateDescriptor<String, String> bcStateDesc = new MapStateDescriptor<>("bcState", String.class, String.class);
        // 1.把controlStream 做成广播流
        BroadcastStream<String> broadcast = controlStream.broadcast(bcStateDesc);
        // 2.数据流和广播流进行connect
        dataStream.connect(broadcast)
                .process(new BroadcastProcessFunction<String, String, String>() {
                    //处理数据流的数据
                    @Override
                    public void processElement(String value,
                                               ReadOnlyContext ctx,
                                               Collector<String> out) throws Exception {
                        //把广播流中的读取配置信息
                        ReadOnlyBroadcastState<String, String> state = ctx.getBroadcastState(bcStateDesc);
                        String s = state.get("switch");
                        if ("1".equals(s)) {
                            out.collect("对" + value + "的处理使用: 1 号逻辑");
                        }else if ("2".equals(s)){
                            out.collect("对" + value + "的处理使用: 2 号逻辑");
                        }else {
                            out.collect("对" + value + "的处理使用: default 号逻辑");
                        }

                    }

                    // 处理广播流的数据
                    @Override
                    public void processBroadcastElement(String value,
                                                        Context ctx,
                                                        Collector<String> out) throws Exception {

                        //把广播流中的数据，写入到广播状态中
                        BroadcastState<String, String> state = ctx.getBroadcastState(bcStateDesc);
                        //把控制信息写入到广播状态
                        state.put("switch",value);

                    }
                })
                .print();

        env.execute();
    }
/*
*       两个流：
*           用其中一个流（配置流） 来控制另外一个（数据流）流的数据 的处理方式
*           1.数据流，控制流
*           2.把控制流做成广播流
*           3.把广播流的数据放入广播状态
*
*
* */


}
